package tjs.study.notes.queue.consumeA.kafka;

import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import tjs.study.notes.queue.consumeA.base.MsgBase;
import tjs.study.notes.queue.consumeA.kafka.myStream.MySource;

import java.util.Map;

//消息接受端，stream给我们提供了Sink,Sink源码里面是绑定input通道的，要跟我们配置文件的imput关联的。
@EnableBinding({Sink.class, MySource.class})
public class RecieveService {

    // 头部headers中的type属性满足条件才消费
    @StreamListener(target = MySource.INPUT, condition = "headers['type']=='testHeader'")
    public void testHeader(@Payload MsgBase message) {
        System.out.println("testHeader: " + message);
    }

    @StreamListener(MySource.INPUT)
    public void recieveMy(Object payload, @Headers Map<String, Object> headers) {
        System.out.println("my: " + payload);
        System.out.println("my: " + headers);
    }

    @StreamListener(Sink.INPUT)
    public void recieve(Object payload) {
        System.out.println("default: " + payload);
    }
}